package com.shujia.flink.window

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo4ProcessFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))


    //将同一个单词分到同一个窗口中
    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = kvDS
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))


    /**
      * process: flink 底层api ,可以操作flink的时间，事件，状态
      *
      */
    //统计单词额数量，返回，单词，窗口结束时间，单词的数量

    val countDS: DataStream[(String, Long, Int)] = windowDS.process(new ProcessWindowFunction[(String, Int), (String, Long, Int), String, TimeWindow] {

      /**
        * process: 每一个key的每一个窗口执行一次process方法
        *
        * @param key      ： key
        * @param context  : 上下文对象，可以获取到窗口的开始和结束时间
        * @param elements ：一个key在窗口内所有的数据，是一个迭代器
        * @param out      ：用于将数据发送到下游
        */
      override def process(key: String,
                           context: Context,
                           elements: Iterable[(String, Int)],
                           out: Collector[(String, Long, Int)]): Unit = {

        //一个单词在一个窗口内的数量
        val count: Int = elements.size

        //获取窗口的结束时间
        val winEndTime: Long = context.window.getEnd

        //将数据发送到下游
        out.collect((key, winEndTime, count))
      }
    })


    countDS.print()


    env.execute()

  }

}
